package com.study.spark.scala.rdd

import org.apache.spark.{SparkConf, SparkContext}

/**
  * reduce side join
  * 适用于两个join表数据量都很大的情况
  *
  * @author stephen
  * @create 2019-03-17 13:13
  * @since 1.0.0
  */
object ReduceSideJoinDemo {
  def main(args: Array[String]) {
    val conf = new SparkConf()
    conf.setAppName("Reduce Side Join Demo")
    conf.setMaster("local[3]")
    conf.set("spark.shuffle.manager", "sort");
    val sc = new SparkContext(conf)

    val table1 = sc.parallelize(List("k1,v11", "k2,v12", "k3,v13"))
    val table2 = sc.parallelize(List("k1,v21", "k4,v24", "k3,v23"))
    //table1 and table 2 are both very large
    val pairs1 = table1.map { x =>
      val pos = x.indexOf(',')
      (x.substring(0, pos), x.substring(pos + 1))
    }

    val pairs2 = table2.map { x =>
      val pos = x.indexOf(',')
      (x.substring(0, pos), x.substring(pos + 1))
    }
    val result = pairs1.join(pairs2)
    //save result to local file or HDFS
    result.saveAsTextFile("/your/path")
  }
}
